package glory.xxd

import org.apache.activemq.ActiveMQConnectionFactory
import org.apache.commons.logging.LogFactory
import org.json.JSONObject
import tsCIM.ResultCode
import tsCIM.modelForInterface.ResultModel

import javax.jms.*

/**
 * Created by Administrator on 2018-01-23.
 */
class ConsumerUtil {

    private final static logger = LogFactory.getLog(this)

    private static final String USERNAME = "admin";
    private static final String PASSWORD = "admin";
    private static final String IP = "tcp://127.0.0.1:61616";
//    private static final String IP = "tcp://10.16.100.19:61616";
    //private static final String BROKEURL = "tcp://127.0.0.1:61616"
    //failover:(tcp://10.10.10.44:61620)?randomize=false
    private static final String BROKEURL = "failover:(${IP})?randomize=false"
    private static final String BROKEURL_QUE = IP

    //private static final String BROKEURL =  "tcp://192.168.33.240:61624"

    //public static newMsgMap = [:]

    /**
     * glory.xxd 接收MQ queue消息
     * 2019-03-16
     * @param msgId mq msgid
     * @param returnMqMsgName
     * @param testFlag 测试返回
     * @return
     */
    public static   getQueueMsg(String msgId, String returnMqMsgName,testFlag) {

        def result = ResultModel.getResultModel()

        if(testFlag){
            result.put(ResultModel.Data,testFlag)
            return result
        }

        Connection connection
        Session session
        try {
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ConsumerUtil.USERNAME, ConsumerUtil.PASSWORD, ConsumerUtil.BROKEURL_QUE);
            connection = connectionFactory.createConnection();
            connection.start();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue(returnMqMsgName);
            MessageConsumer messageConsumer = session.createConsumer(destination, "JMSCorrelationID='${msgId}'");
            //println messageConsumer.getMessageSelector();//延时20分钟
            TextMessage textMessage = (TextMessage) messageConsumer.receive(15000);
            if (textMessage) {
                result.put(ResultModel.Data, textMessage.getText())
            }
            logger.info("---------------${new Date()} Get Queue:${ConsumerUtil.BROKEURL_QUE} Sucess!MSGID:${msgId}---------------")
            logger.info("---------------${result.get(ResultModel.Data)?:"Receive MQ:${returnMqMsgName} has not msg!! "}---------------")
        } catch (Exception e) {
            result.put(ResultModel.Code,ResultCode.RESPONSE_MQ_ERROR.code)
            result.put(ResultModel.Msg,ResultCode.RESPONSE_MQ_ERROR.msg)
            logger.error( "---------------${new Date()} Get Queue:${this.BROKEURL_QUE} Erorr!MSGID:${msgId} MSG_NAME:${returnMqMsgName}---------------")
            logger.error(e.getMessage())
            e.printStackTrace()
        }finally{
            if(session){
                session.close()
                session = null
            }
            if(connection){
                connection.close();
                connection = null
            }
            return result;
        }
    }

    /**
     * 专门用于监听topic信息
     * @param msgName
     * @parma max 消息条数
     * @param newMsgMap
     */
    public static void getTopicMsg(String msgName,int max, Map<String, LinkedList> newMsgMap) {

        try{
            //创建一个ConnectionFactory对象连接MQ服务器
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ConsumerUtil.USERNAME, ConsumerUtil.PASSWORD, ConsumerUtil.BROKEURL);
            //创建一个连接对象
            Connection connection = connectionFactory.createConnection();
            //开启连接
            connection.start();
            //使用Connection对象创建一个Session对象
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //创建一个Destination对象。topic对象
            Topic topic = session.createTopic(msgName);
            //使用Session对象创建一个消费者对象。
            MessageConsumer consumer = session.createConsumer(topic);

            consumer.setMessageListener(new MessageListener() {
                @Override
                public void onMessage(Message message) {
                    if (message) {
                        TextMessage textMessage = (TextMessage) message;
                        String xml = textMessage.getText()
                        String msgId = message.getJMSMessageID()

                        JSONObject json =  Utils.getJsonObjectByXMLAndNode(xml,'Request')
                        def eqpId = json.EQPID ?: ""
                        json.put("msgId",msgId)

                        if (eqpId) {
                            LinkedList list = newMsgMap?."${eqpId}"

                            if (list) {
                                list.addLast(json)
                                if (list.size() > max) {
                                    list.removeFirst()
                                }

                            } else {
                                list = LinkedList.newInstance()
                                list.addFirst(json)
                                newMsgMap.put(eqpId, list)
                            }
                        }

                        logger.info( "Real-time news ${ConsumerUtil.BROKEURL}getTopic Sucess!MSGID:${msgName}---------------")
                        logger.info("Real-time news: ${xml?:'has no message...........'}")

                    }else{
                        logger.info( "Real-time news${new Date()}${ConsumerUtil.BROKEURL}getTopic no message!MSGID:${msgName}-----------")
                    }
                }
            });
        }catch (Exception e){
            logger.error("MessageListener for Topic:${msgName} Error......................")
            logger.error(e.getMessage())
            e.printStackTrace()
        }
    }
}